package com.flink.examples;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.util.PropertiesUtil;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;

public class CkSinkUtil {
    static Properties prop = PropertiesUtil.getProp();
    /**
     * 生成数据载入clickhouse时，所需的insertSql
     * @param clazz
     * @return
     * @param <T>
     */
    public static <T> String generateInsertSql(Class<T> clazz, String tableName) {
        Field[] fields = clazz.getDeclaredFields();
        List<String> fieldNames = new ArrayList<>();
        List<String> qMarks = new ArrayList<>();
        for (Field field : fields) {
            String fieldName = field.getName();
            if (StringUtils.equals(fieldName, "desc")) {
                fieldNames.add("description");
            } else if (StringUtils.equals(fieldName, "result")) {
                fieldNames.add("result_info");
            } else if (StringUtils.equals(fieldName, "clazz")) {
                fieldNames.add("class");
            } else {
                fieldNames.add(field.getName());
            }
            qMarks.add("?");
        }
        StringBuffer sb = new StringBuffer();
        sb.append("INSERT INTO ");
        sb.append(tableName);
        sb.append(" (");
        sb.append(StringUtils.join(fieldNames, ","));
        sb.append(")\n");
        sb.append("VALUES (");
        sb.append(StringUtils.join(qMarks, ","));
        sb.append(")");
        return sb.toString();
    }

    public static <T> JdbcStatementBuilder getJdbcStatementBuilder(Class<T> clazz) {
        JdbcStatementBuilder jdbcStatementBuilder =
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement,
                                       T t) throws SQLException {
                        try {
                            Field[] fields = clazz.getDeclaredFields();
                            Integer count = 1;
                            for (Field field : fields) {
                                String fieldName = field.getName();
                                String fieldType = field.getType().toString();
                                if (StringUtils.equals(fieldType, "class java.lang.String")) {
                                    Method getMethod = findGetMethodByFieldName(clazz, fieldName);
                                    preparedStatement.setString(count, (String) getMethod.invoke(t));
                                } else if (StringUtils.equals(fieldType, "class java.lang.Long")) {
                                    Method getMethod = findGetMethodByFieldName(clazz, fieldName);
                                    preparedStatement.setLong(count, (Long) getMethod.invoke(t));
                                } else if (StringUtils.equals(fieldType, "class java.util.Date")) {
                                    Method getMethod = findGetMethodByFieldName(clazz, fieldName);
                                    preparedStatement.setTimestamp(count,
                                            new Timestamp(((Date) getMethod.invoke(t)).getTime()));
                                } else if (StringUtils.equals(fieldType, "class java.lang.Integer")) {
                                    Method getMethod = findGetMethodByFieldName(clazz, fieldName);
                                    preparedStatement.setInt(count, (int) getMethod.invoke(t));
                                } else {
                                    // do nothing
                                }
                                count += 1;
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                };
        return jdbcStatementBuilder;
    }

    public static JdbcExecutionOptions getJdbcExecutionOptions() {
        Integer batchSize =
                Integer.valueOf(prop.getProperty("clickhouse.insert.batch.size"));
        Integer maxRetries =
                Integer.valueOf(prop.getProperty("clickhouse.insert.maxretries"));
        Long batchInterval =
                Long.valueOf(prop.getProperty("clickhouse.insert.batch.intervalms"));

        JdbcExecutionOptions jdbcExecutionOptions = new JdbcExecutionOptions.Builder()
                .withBatchSize(batchSize)
                .withMaxRetries(maxRetries)
                .withBatchIntervalMs(batchInterval)
                .build();
        return jdbcExecutionOptions;
    }

    public static JdbcConnectionOptions getJdbcConnectionOptions() {
        JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(prop.getProperty("clickhouse.jdbc.drivername"))
                .withUrl(prop.getProperty("clickhouse.jdbc.url"))
                .withUsername(prop.getProperty("clickhouse.user"))
                .withPassword(prop.getProperty("clickhouse.password"))
                .build();
        return jdbcConnectionOptions;
    }

    private static Method findGetMethodByFieldName(Class<?> clazz,
                                                   String fieldName) throws NoSuchMethodException {
        String getMethodName = "get"
                + fieldName.substring(0, 1).toUpperCase()
                + fieldName.substring(1);
        return clazz.getMethod(getMethodName);
    }
}
